package com.pw.study.flink.chapter3;

import com.pw.study.flink.entities.UserBehavior;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class UserBehavior_PV {

    public static void main(String[] args) {
        dealPv();
    }

    private static void dealPv() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        env.readTextFile("data/file/UserBehavior.csv").map(line -> {
            String[] sp = line.split(",");
            return new UserBehavior(
                    Long.valueOf(sp[0]), Long.valueOf(sp[1]), Integer.valueOf(sp[2]), sp[3], Long.valueOf(sp[4])
            );
        }).filter(x -> x.getBehavior().equals("pv")).map(x -> Tuple2.of("Pv", 1L)
        ).returns(Types.TUPLE(Types.STRING,Types.LONG)).keyBy(t -> t.f0).sum(1).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
